package metadata_manager

import (
	"errors"
	"gitee.com/zaper/kafka-service/service/metadata/model"
	"gitee.com/zaper/private-logger"
	"github.com/segmentio/kafka-go"
)

func (metadataManager *MetadataManager) GetAllTopics() []kafka.Topic {
	meta, err := metadataManager.client.Metadata(metadataManager.ctx, &kafka.MetadataRequest{})
	if err != nil || meta == nil {
		return nil
	}
	return meta.Topics
}

func (metadataManager *MetadataManager) GetTopicByName(topicName string) *kafka.Topic {
	topics := metadataManager.GetAllTopics()
	if topics == nil {
		return nil
	}
	for _, t := range topics {
		if t.Name == topicName {
			return &t
		}
	}
	return nil
}

func (metadataManager *MetadataManager) CreateTopic(model *model.TopicModel) error {
	var topic *kafka.Topic
	if topic = metadataManager.GetTopicByName(model.TopName); topic != nil {
		logger.Infof("topic is exit")
		return nil
	}
	if _, err := metadataManager.client.CreateTopics(metadataManager.ctx, &kafka.CreateTopicsRequest{
		Topics: []kafka.TopicConfig{
			{
				Topic:             model.TopName,
				NumPartitions:     model.NumPartition,
				ReplicationFactor: model.NumReplications, // 单副本（生产环境建议 >= 3）
				ConfigEntries:     model.CfgEntry,
			},
		},
	}); err != nil {
		return err
	}
	if topic = metadataManager.GetTopicByName(model.TopName); topic == nil {
		return nil
	}
	logger.Infof("Current partitions num is %v", topic)
	return nil
}

func (metadataManager *MetadataManager) DeleteTopic(model *model.TopicModel) error {
	topic := metadataManager.GetTopicByName(model.TopName)
	if topic == nil {
		return nil
	}
	resp, err := metadataManager.client.DeleteTopics(metadataManager.ctx, &kafka.DeleteTopicsRequest{
		Topics: []string{model.TopName},
	})
	if err != nil || resp.Errors[model.TopName] != nil {
		logger.Info("Delete topic err %v, resp err is %v", err, resp.Errors[model.TopName])
		return errors.New("delete topic err")
	}
	logger.Info("Delete topic success!")
	return nil
}
